package com.github.xiaolyuh;


import com.github.xiaolyuh.configcenter.ConfigCenter;
import com.github.xiaolyuh.constant.CacheConstant;
import com.github.xiaolyuh.entity.MqMessageModel;
import com.github.xiaolyuh.entity.RedisName;
import com.github.xiaolyuh.enums.AckStatusEnum;
import com.github.xiaolyuh.enums.CacheCodeEnum;
import com.github.xiaolyuh.enums.ModelEnum;
import com.github.xiaolyuh.enums.MqMessageTypeEnum;
import com.github.xiaolyuh.exception.CacheException;
import com.github.xiaolyuh.exception.PromotionMonitorException;
import com.github.xiaolyuh.manager.MessageStorage;
import com.github.xiaolyuh.utils.FastJSONHelper;
import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.NumberUtils;

import java.util.List;
import java.util.Objects;

/**
 * @author yizuchao
 */
@Component
public class PubSubWithAckRedisManager{

    private static final Logger logger= LoggerFactory.getLogger(PubSubWithAckRedisManager.class);



    public PubSubWithAckRedisManager() {
        this.baseRedisManager = new BaseRedisManager();
    }

    /**
     * 发送redis消息
     * @param siteCode
     * @param typeEnum
     * @param content
     */
    public void sendMsg(String siteCode,MqMessageTypeEnum typeEnum,Object content,MessageStorage messageStorage) throws CacheException {
        String contentStr= FastJSONHelper.serialize(content);
        String msgId=String.valueOf(IDGenerator.getInstance().getId());
        try{
            messageStorage.store(siteCode,msgId,contentStr,typeEnum);
        }catch (Exception e){
            String failMsg=String.format("保存ack消息失败。type= %s content= %s",typeEnum.getType(),contentStr);
            logger.error(failMsg, PromotionMonitorException.of(PromotionMonitorException.DEFAULT_CODE, failMsg));
        }
        MqMessageModel mqMessage=new MqMessageModel(siteCode,msgId,typeEnum.getType(),contentStr);
        String serializeMsg=FastJSONHelper.serialize(mqMessage);
        if(!baseRedisManager.publish(getRedisName(siteCode),typeEnum.getDestination(),serializeMsg)){
            throw new CacheException(CacheCodeEnum.SEND_MSG_ERROR.getCode(),String.format("[cache] 消息发送失败。type= %s content= %s",
                    typeEnum.getType(),serializeMsg));
        }
    }


    /**
     * ack消息
     * @param siteCode
     * @param msgId
     * @return
     */
    public boolean ackMsg(String siteCode,String msgId,String ip){
        String key = CacheKeyBuilder.buildRedisMsgAckKey(siteCode, msgId);
        try{
            baseRedisManager.sadd(getRedisName(siteCode),key,new String[]{ip}, CacheConstant.CACHE_DEFAULT_TIMEOUT);
            return true;
        }catch (Exception e){
            logger.warn("ack 消息失败。msgId={} ip={}",msgId,ip);
        }
        return false;
    }

    /**
     * 订阅
     * @param siteCode
     * @param jedisPubSub
     * @param channel
     * @return
     */
    public boolean subscribe(String siteCode,JedisPubSub jedisPubSub, String channel){
        return baseRedisManager.subscribe(getRedisName(siteCode),jedisPubSub,channel);
    }

    /**
     * 判断消息是否确认50010
     * @param siteCode
     * @param pageSize
     * @return
     */
    public void checkAndAck(String siteCode,Integer pageSize,MessageStorage messageStorage){
        Integer pageIndex= NumberUtils.INTEGER_ZERO;
        while (true){
            try{
                List<MqMessageModel> mqMessages = messageStorage.getNewestOfToBeAck(siteCode,pageIndex,pageSize);
                if(CollectionUtils.isEmpty(mqMessages)){
                    break;
                }
                List<String> ackSuccessMsgIds= Lists.newArrayList();
                List<String> ackFailMsgIds=Lists.newArrayList();
                List<String> incrRetriedTimesMsgIds=Lists.newArrayList();
                for (MqMessageModel mqMessageModel:mqMessages){
                    //已经确认的消息id
                    if(checkIfAcked(siteCode,mqMessageModel.getMsgId())){
                        ackSuccessMsgIds.add(mqMessageModel.getMsgId());
                        continue;
                    }

                    Integer maxRetryTimes= Objects.isNull(mqMessageModel.getMaxRetryTimes())?NumberUtils.INTEGER_ONE:mqMessageModel.getMaxRetryTimes();
                    Integer retiedTimes=Objects.isNull(mqMessageModel.getRetriedTimes())?NumberUtils.INTEGER_ZERO:mqMessageModel.getRetriedTimes();
                    if(Math.addExact(retiedTimes,NumberUtils.INTEGER_ONE)>=maxRetryTimes){
                        String failMsg=String.format("msgId:%s ack失败",mqMessageModel.getMsgId());
                        logger.error(failMsg, PromotionMonitorException.of(PromotionMonitorException.DEFAULT_CODE, failMsg));
                        ackFailMsgIds.add(mqMessageModel.getMsgId());
                        continue;
                    }else{
                        reSendMsg(mqMessageModel.getSiteCode(),mqMessageModel.getMsgId(),
                                MqMessageTypeEnum.valueOf(mqMessageModel.getType()),mqMessageModel.getContent());
                        incrRetriedTimesMsgIds.add(mqMessageModel.getMsgId());
                    }
                }

                //更新消息状态
                messageStorage.updateAckResult(ackSuccessMsgIds, AckStatusEnum.ACK_SUCCESS);
                messageStorage.updateAckResult(ackFailMsgIds,AckStatusEnum.ACK_FAIL);
                messageStorage.incrRetriedTimes(incrRetriedTimesMsgIds);
            }catch (Exception e){
                logger.error("检查消息是否确认执行失败，pageIndex:{}",pageIndex,e);
            }
        }
    }

    private boolean checkIfAcked(String siteCode,String msgId){
        Integer ackSize=baseRedisManager.scard(getRedisName(siteCode),CacheKeyBuilder.buildRedisMsgAckKey(siteCode, msgId));
        Integer soaIpNum= ConfigCenter.getSoaIpNum();
        if(ackSize>soaIpNum){
            String warnMsg=String.format("消息id：%s 消息确认消息机器数大于配置中心的机器数，" +
                    "请确认配置中心的机器数量是否正确或者是否在重启机器",msgId);
            logger.error(warnMsg,PromotionMonitorException.of(PromotionMonitorException.DEFAULT_CODE,warnMsg));
        }
        return soaIpNum<=ackSize;
    }

    protected RedisName getRedisName(String siteCode){
        return new RedisName(siteCode, ModelEnum.ACTIVITY);
    }

    /**
     * 重发消息
     * @return
     */
    private boolean reSendMsg(String siteCode,String msgId,MqMessageTypeEnum typeEnum,String content) throws CacheException{
        MqMessageModel mqMessageModel=new MqMessageModel(siteCode,msgId,typeEnum.getType(),content);
        String serializeMsg=FastJSONHelper.serialize(mqMessageModel);
        if(!baseRedisManager.publish(getRedisName(mqMessageModel.getSiteCode()),typeEnum.getDestination(),serializeMsg)){
            throw new CacheException(CacheCodeEnum.SEND_MSG_ERROR.getCode(),String.format("[cache] 消息重发送失败。type= %s content= %s",
                    typeEnum.getType(),serializeMsg));
        }
        return true;
    }
}
